package edu.utexas.mapreduce;

import java.net.URL;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * Local copy of the job state
 */
class JobStateLocal implements JobState {

	final int jobID;
	final String outputDir;
	final String emailAddress;
	boolean mapComplete;
	boolean reduceComplete;
	boolean notificationSent;
	List<MapTask> mapTasks;
	List<ReduceTask> reduceTasks;
	
	JobStateLocal(int id, String outputDir, List<InputBlock> input, String emailAddress) {
		this.jobID = id;
		this.outputDir = outputDir;
		this.emailAddress = emailAddress;
		this.mapComplete = false;
		this.reduceComplete = false;
		this.notificationSent = false;
		this.mapTasks = new ArrayList<MapTask>(input.size());
		this.reduceTasks = new ArrayList<ReduceTask>(Constants.NUMBER_OF_REDUCE_TASKS);
		
		// initialize lists
		for (int i = 0; i < input.size(); i++) {
			mapTasks.add(new MapTask(i, input.get(i)));
		}
		
		for (int i = 0; i < Constants.NUMBER_OF_REDUCE_TASKS; i++) {
			reduceTasks.add(new ReduceTask(i));
		}
	}
	
	public String getOutputDir() {
		return outputDir;
	}
	
	public int getJobID() {
		return jobID;
	}
	
	public List<MapTask> getMapTasks() {
		return mapTasks;
	}
	
	public List<ReduceTask> getReduceTasks() {
		return reduceTasks;
	}

	public MapTask reserveIdleMapTask(String owner) {
		MapTask map = findMapTask(TaskState.IDLE);
		if (map != null) {
			map.setOwner(owner);
			map.setState(TaskState.IN_PROGRESS);
		}
		return map;
	}
	
	public ReduceTask reserveIdleReduceTask(String owner) {
		
		// cannot schedule a reduce task if map tasks are pending
		if (!mapComplete) {
			return null;
		}
		
		ReduceTask reduce = findReduceTask(TaskState.IDLE);
		if (reduce != null) {
			reduce.setOwner(owner);
			reduce.setState(TaskState.IN_PROGRESS);
		}
		return reduce;
	}
	
	public void completeReduceTask(String owner, ReduceTask task) {
		if (task.getOwner().equals(owner)) {
			task.setState(TaskState.COMPLETE);
		}
	}
	
	private MapTask findMapTask(TaskState state) {
		for(MapTask map : mapTasks) {
			if (map.getState() == state) {
				return map;
			}
		}
		return null;
	}

	private ReduceTask findReduceTask(TaskState state) {
		for(ReduceTask reduce : reduceTasks) {
			if (reduce.getState() == state) {
				return reduce;
			}
		}
		return null;
	}
	
	public boolean mapTasksComplete(Set<String> activeProcessSet) {
		if (activeProcessSet != null) {
			getActiveProcessSet(activeProcessSet);
		}
		return mapComplete;
	}
	
	public boolean reduceTasksComplete(Set<String> activeProcessSet) {
		
		if (activeProcessSet != null) {
			getActiveProcessSet(activeProcessSet);
		}
		
		// reduce cannot complete if there are pending
		// map tasks
		if (!mapComplete) {
			return false;
		}
		
		for(ReduceTask reduce : reduceTasks) {
			if (reduce.getState() != TaskState.COMPLETE) {
				return false;
			}
		}
		return true;
	}
	
	public void commitMapResults(String owner, Map<Integer, URL> results) {
		
		boolean allTasksComplete = true;
		for(MapTask mapTask : mapTasks) {
			if (mapTask.getOwner().equals(owner)) {
				mapTask.setState(TaskState.COMPLETE);
			}
			
			allTasksComplete &= (mapTask.getState() == TaskState.COMPLETE);
		}
		
		mapComplete = allTasksComplete;
		
		for(Map.Entry<Integer, URL> entry: results.entrySet()) {
			ReduceTask task = reduceTasks.get(entry.getKey().intValue());
			task.setState(TaskState.IDLE);
			task.addMapResults(entry.getValue());
		}
	}
	
	public void recoverMapTaskFailure(String owner) {
		
	}
	
	public void recoverFromProcessFailure(String failedProcess) {
		
		// Only revert tasks in the IN_PROGRESS state.  COMPLETED task results
		// are committed to the distributed file system
		for(ReduceTask reduceTask : reduceTasks) {
			if (reduceTask.getOwner().equals(failedProcess) && reduceTask.getState() == TaskState.IN_PROGRESS) {
				reduceTask.setOwner("");
				reduceTask.setState(TaskState.IDLE);
			}
		}
		
		// clear map results from idle reduce tasks
		for(ReduceTask reduceTask : reduceTasks) {
			if (reduceTask.getState() == TaskState.IDLE) {
				Iterator<URL> it = reduceTask.getMapResults().iterator();
				while(it.hasNext()) {
					URL url = it.next();
					final int port = url.getPort();
					final String process = url.getHost() + ((port != -1) ? (":" + port) : "");
					if (process.equals(failedProcess)) {
						it.remove();
					}
				}
			}
		}
		
		// recover in-progress and completed map tasks
		for(MapTask mapTask : mapTasks) {
			if (mapTask.getOwner().equals(failedProcess)) {
				mapTask.setOwner("");
				mapTask.setState(TaskState.IDLE);
				mapComplete = false;
			}
		}
	}

	public void close() {
		// no-op
	}
	
	public String toString() {
		StringBuilder sb = new StringBuilder();
		sb.append("jobID    : " + jobID + "\n");
		sb.append("outputDir: " + outputDir + "\n");
		sb.append("map tasks:\n");
		
		for(MapTask task : mapTasks) {
			sb.append(" id   : " + task.getID() + "\n");
			sb.append(" state: " + task.getState().toString() + "\n");
			sb.append(" owner: " + task.getOwner() + "\n\n");
		}
		sb.append("reduce tasks:\n");
		for(ReduceTask task : reduceTasks) {
			sb.append(" id   : " + task.getID() + "\n");
			sb.append(" state: " + task.getState().toString() + "\n");
			sb.append(" owner: " + task.getOwner() + "\n");
			sb.append(" map  : ");
			for(URL url: task.getMapResults()) {
				sb.append(url + "\n        ");
			}
			sb.append("\n");
		}
		
		return sb.toString();
	}

	public void markMapTaskInProgress(int id, String owner) {
		Task t = mapTasks.get(id);
		t.setState(TaskState.IN_PROGRESS);
		t.setOwner(owner);
	}

	public void markReduceTaskIdle(int id, String owner) {
		Task t = reduceTasks.get(id);
		t.setState(TaskState.IDLE);
		t.setOwner(owner);
	}
	
	public void markReduceTaskInProgress(int id, String owner) {
		Task t = reduceTasks.get(id);
		t.setState(TaskState.IN_PROGRESS);
		t.setOwner(owner);
	}

	public void markReduceTaskComplete(int id, String owner) {
		Task t = reduceTasks.get(id);
		t.setState(TaskState.COMPLETE);
		t.setOwner(owner);
	}
	
	void getActiveProcessSet(Set<String> activeProcessSet) {
		
		for (Task task : mapTasks) {
			if (task.getState() == TaskState.IN_PROGRESS) {
				activeProcessSet.add(task.getOwner());
			}
		}
		for (Task task : reduceTasks) {
			if (task.getState() == TaskState.IN_PROGRESS) {
				activeProcessSet.add(task.getOwner());
			}
		}
	}

	public void markNotificationSent() {
		notificationSent = true;
	}

	public void sendNotification() {
		
		if (!notificationSent && emailAddress != null) {
			// send the notification
			System.out.println("send notification to: " + emailAddress + " DISABLED!!! (see JobStateLocal.java:line273");
			
			// TODO - need to set email server parameter for the server 
			//final String smtpHost = "";
			//final String serverEmail = "";
			//final String username = "";
			//final String password = "";
			//Notification notification = new Notification(smtpHost, username, password);
			//notification.sendEmail(serverEmail, emailAddress, "JOB " + jobID + " DONE", this.toString());
		}
		notificationSent = true;
	}


}
